/*
 * This file and its contents are licensed under the Timescale License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-TIMESCALE for a copy of the license.
 */
#include <postgres.h>
#include <utils/builtins.h>
#include <utils/memutils.h>
#include <utils/palloc.h>
#include <utils/snapmgr.h>
#include <utils/tuplestore.h>
#include <nodes/makefuncs.h>
#include <nodes/memnodes.h>
#include <storage/lockdefs.h>
#include <access/htup_details.h>
#include <access/htup.h>
#include <access/xact.h>
#include <miscadmin.h>

#include <extension.h>
#include <fmgr.h>
#include <funcapi.h>
#include <parser/parse_func.h>
#include <scanner.h>
#include <scan_iterator.h>
#include <utils.h>
#include <time_utils.h>
#include <time_bucket.h>
#include <hypertable_cache.h>

#include "compat/compat.h"
#include "remote/dist_commands.h"
#include "ts_catalog/catalog.h"
#include "ts_catalog/continuous_agg.h"
#include "continuous_aggs/materialize.h"
#include "data_node.h"
#include "deparse.h"
#include "invalidation.h"
#include "refresh.h"

/*
 * Invalidation processing for continuous aggregates.
 *
 * Invalidations track the regions/intervals [start, end] of a continuous
 * aggregate that are out-of-date relative to the source hypertable on top of
 * which the aggregate is defined. When a continuous aggregate is out-of-date
 * across one or more regions, it can be refreshed with a window covering
 * those regions in order to bring it up-to-date with the source data again.
 *
 * Invalidations are generated by mutations on the source hypertable (INSERT,
 * DELETE, UPDATE, TRUNCATE, drop_chunks, etc.) and are initially written to a
 * hypertable invalidation log [hypertable_id, start, end].
 *
 * When a continuous aggregate is refreshed, invalidations are moved from the
 * hypertable invalidation log to a continuous aggregate invalidation log,
 * where each original entry creates one new entry per continuous aggregate
 * [cagg_id, start, end]. Thus, if one continuous aggregate is refreshed but
 * not others, then only the invalidations for the refreshed aggregate are
 * processed.
 *
 * Simplified, invalidations move through the following stages:
 *
 * insert_trigger => hypertable_inval_log => cagg_inval_log => refreshing
 *
 * Thus, invalidations are generated by mutations and are processed and used
 * as input for refreshing the a continuous aggregate.
 *
 * Invalidations can overlap or be duplicates. Therefore, invalidations are
 * merged during processing to reduce the number of entries in the logs. This
 * typically happens during a refresh of a continuous aggregate, which also
 * cuts invalidations along the refresh window. The cutting will leave some
 * parts of entries in the invalidation log while the entries that fall
 * within the refresh window are stored in an invalidation store and used for
 * refreshing:
 *
 *       |-------------|    refresh window
 *
 *   |-----|  |---| |----|  invalidations
 *
 *             =>
 *
 *   |---|             |--| invalidations that remain in the log
 *
 *       |-|  |---| |--|    invalidations that are used for refreshing
 *
 * The invalidation store will spill to disk in case of many invalidations so
 * it won't blow up memory usage. If there are no invalidations in the store
 * after processing, then the continuous aggregate is up-to-date in the region
 * defined by the refresh window.
 */

typedef struct CaggInvalidationState
{
	int32 mat_hypertable_id;
	int32 raw_hypertable_id;
	Oid dimtype; /* The type of the underlying hypertable's time dimension
				  * that is bucketed */
	MemoryContext per_tuple_mctx;
	Relation cagg_log_rel;
	Snapshot snapshot;
	Tuplestorestate *invalidations;
	const CaggsInfo *all_caggs;
	int64 bucket_width;
	/* bucket_function is NULL unless bucket_width == BUCKET_WIDTH_VARIABLE */
	const ContinuousAggsBucketFunction *bucket_function;
} CaggInvalidationState;

typedef enum LogType
{
	LOG_HYPER,
	LOG_CAGG,
} LogType;

static Relation
open_invalidation_log(LogType type, LOCKMODE lockmode)
{
	static const CatalogTable logmappings[] = {
		[LOG_HYPER] = CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG,
		[LOG_CAGG] = CONTINUOUS_AGGS_MATERIALIZATION_INVALIDATION_LOG,
	};
	Catalog *catalog = ts_catalog_get();
	Oid relid = catalog_get_table_id(catalog, logmappings[type]);

	return table_open(relid, lockmode);
}

static void
hypertable_invalidation_scan_init(ScanIterator *iterator, int32 hyper_id, LOCKMODE lockmode)
{
	*iterator = ts_scan_iterator_create(CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG,
										lockmode,
										CurrentMemoryContext);
	iterator->ctx.index = catalog_get_index(ts_catalog_get(),
											CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG,
											CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG_IDX);
	ts_scan_iterator_scan_key_init(
		iterator,
		Anum_continuous_aggs_hypertable_invalidation_log_idx_hypertable_id,
		BTEqualStrategyNumber,
		F_INT4EQ,
		Int32GetDatum(hyper_id));
}

static HeapTuple
create_invalidation_tup(const TupleDesc tupdesc, int32 cagg_hyper_id, int64 start, int64 end)
{
	Datum values[Natts_continuous_aggs_materialization_invalidation_log] = { 0 };
	bool isnull[Natts_continuous_aggs_materialization_invalidation_log] = { false };

	values[AttrNumberGetAttrOffset(
		Anum_continuous_aggs_materialization_invalidation_log_materialization_id)] =
		Int32GetDatum(cagg_hyper_id);
	values[AttrNumberGetAttrOffset(
		Anum_continuous_aggs_materialization_invalidation_log_lowest_modified_value)] =
		Int64GetDatum(start);
	values[AttrNumberGetAttrOffset(
		Anum_continuous_aggs_materialization_invalidation_log_greatest_modified_value)] =
		Int64GetDatum(end);

	return heap_form_tuple(tupdesc, values, isnull);
}

/*
 * Add an entry to the continuous aggregate invalidation log.
 */
void
invalidation_cagg_log_add_entry(int32 cagg_hyper_id, int64 start, int64 end)
{
	Relation rel = open_invalidation_log(LOG_CAGG, RowExclusiveLock);
	CatalogSecurityContext sec_ctx;
	HeapTuple tuple;

	Assert(start <= end);
	tuple = create_invalidation_tup(RelationGetDescr(rel), cagg_hyper_id, start, end);
	ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
	ts_catalog_insert_only(rel, tuple);
	ts_catalog_restore_user(&sec_ctx);
	heap_freetuple(tuple);
	table_close(rel, NoLock);
}

/**
 * Adds a materialization invalidation log entry to the local data node.
 *
 * @param mat_hypertable_id The hypertable ID of the CAGG materialized hypertable in the Access
 *                          Node.
 * @param start_time The starting time of the materialization invalidation log entry.
 * @param end_time The ending time of the materialization invalidation log entry.
 */
Datum
tsl_invalidation_cagg_log_add_entry(PG_FUNCTION_ARGS)
{
	int32 mat_hypertable_id = PG_GETARG_INT32(0);
	int64 start_time = PG_GETARG_INT64(1);
	int64 end_time = PG_GETARG_INT64(2);

	if (end_time < start_time)
		elog(ERROR, "cannot invalidate cagg, end time should be greater than start time");

	invalidation_cagg_log_add_entry(mat_hypertable_id, start_time, end_time);

	PG_RETURN_VOID();
}

/**
 * Adds a hypertable invalidation log entry to the local data node.
 *
 * @param raw_hypertable_id The hypertable ID of the original distributed hypertable in the Access
 *                          Node.
 * @param start_time The starting time of the materialization invalidation log entry.
 * @param end_time The ending time of the materialization invalidation log entry.
 */
Datum
tsl_invalidation_hyper_log_add_entry(PG_FUNCTION_ARGS)
{
	int32 raw_hypertable_id = PG_GETARG_INT32(0);
	int64 start_time = PG_GETARG_INT64(1);
	int64 end_time = PG_GETARG_INT64(2);

	if (end_time < start_time)
		elog(ERROR, "cannot invalidate hypertable, end time should be greater than start time");

	invalidation_hyper_log_add_entry(raw_hypertable_id, start_time, end_time);

	PG_RETURN_VOID();
}

#define INVALIDATION_CAGG_ADD_ENTRY_NARGS 3
#define INVALIDATION_CAGG_LOG_ADD_ENTRY_FUNCNAME "invalidation_cagg_log_add_entry"
#define INVALIDATION_HYPER_LOG_ADD_ENTRY_FUNCNAME "invalidation_hyper_log_add_entry"

void
remote_invalidation_log_add_entry(const Hypertable *raw_ht,
								  ContinuousAggHypertableStatus caggstatus, int32 entry_id,
								  int64 start, int64 end)
{
	Oid func_oid;
	LOCAL_FCINFO(fcinfo, INVALIDATION_CAGG_ADD_ENTRY_NARGS);
	FmgrInfo flinfo;

	Assert(HypertableIsMaterialization == caggstatus || HypertableIsRawTable == caggstatus);

	static const Oid type_id[INVALIDATION_CAGG_ADD_ENTRY_NARGS] = { INT4OID, INT8OID, INT8OID };
	List *const fqn = list_make2(makeString(INTERNAL_SCHEMA_NAME),
								 makeString((caggstatus == HypertableIsMaterialization) ?
												INVALIDATION_CAGG_LOG_ADD_ENTRY_FUNCNAME :
												INVALIDATION_HYPER_LOG_ADD_ENTRY_FUNCNAME));

	if (!hypertable_is_distributed(raw_ht))
		elog(ERROR, "function was not provided with a valid distributed hypertable");

	func_oid = LookupFuncName(fqn, -1 /* lengthof(type_id) */, type_id, false);
	Assert(OidIsValid(func_oid));

	fmgr_info(func_oid, &flinfo);
	InitFunctionCallInfoData(*fcinfo,
							 &flinfo,
							 INVALIDATION_CAGG_ADD_ENTRY_NARGS,
							 InvalidOid,
							 NULL,
							 NULL);

	FC_NULL(fcinfo, 0) = false;
	FC_ARG(fcinfo, 0) = Int32GetDatum(entry_id);
	FC_NULL(fcinfo, 1) = false;
	FC_ARG(fcinfo, 1) = Int64GetDatum(start);
	FC_NULL(fcinfo, 2) = false;
	FC_ARG(fcinfo, 2) = Int64GetDatum(end);
	/* Check for null result, since caller is clearly not expecting one */
	if (fcinfo->isnull)
		elog(ERROR, "function %u returned NULL", flinfo.fn_oid);

	DistCmdResult *result;
	List *data_node_list = ts_hypertable_get_data_node_name_list(raw_ht);
	result = ts_dist_cmd_invoke_func_call_on_data_nodes(fcinfo, data_node_list);
	if (result)
		ts_dist_cmd_close_response(result);
}

void
invalidation_hyper_log_add_entry(int32 hyper_id, int64 start, int64 end)
{
	Relation rel = open_invalidation_log(LOG_HYPER, RowExclusiveLock);
	CatalogSecurityContext sec_ctx;
	Datum values[Natts_continuous_aggs_hypertable_invalidation_log];
	bool nulls[Natts_continuous_aggs_hypertable_invalidation_log] = { false };

	Assert(start <= end);
	values[AttrNumberGetAttrOffset(
		Anum_continuous_aggs_hypertable_invalidation_log_hypertable_id)] = Int32GetDatum(hyper_id);
	values[AttrNumberGetAttrOffset(
		Anum_continuous_aggs_hypertable_invalidation_log_lowest_modified_value)] =
		Int64GetDatum(start);
	values[AttrNumberGetAttrOffset(
		Anum_continuous_aggs_hypertable_invalidation_log_greatest_modified_value)] =
		Int64GetDatum(end);

	ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
	ts_catalog_insert_values(rel, RelationGetDescr(rel), values, nulls);
	ts_catalog_restore_user(&sec_ctx);
	table_close(rel, NoLock);
	elog(DEBUG1,
		 "hypertable log for hypertable %d added entry [" INT64_FORMAT ", " INT64_FORMAT "]",
		 hyper_id,
		 start,
		 end);
}

/*
 * Invalidate one or more continuous aggregates.
 *
 * Add an invalidation in the given range. The invalidation is added either to
 * the hypertable invalidation log or the continuous aggregate invalidation
 * log depending on the type of the given hypertable. If the hypertable is a
 * "raw" hypertable (i.e., one that has one or more continous aggregates), the
 * entry is added to the hypertable invalidation log and will invalidate all
 * the associated continuous aggregates. If the hypertable is instead an
 * materialized hypertable, the entry is added to the cagg invalidation log
 * and only invalidates the continuous aggregate owning that materialized
 * hypertable.
 */
void
continuous_agg_invalidate_raw_ht(const Hypertable *raw_ht, int64 start, int64 end)
{
	Assert(raw_ht != NULL);

	if (hypertable_is_distributed(raw_ht))
	{
		remote_invalidation_log_add_entry(raw_ht, HypertableIsRawTable, raw_ht->fd.id, start, end);
	}
	else
	{
		invalidation_hyper_log_add_entry(raw_ht->fd.id, start, end);
	}
}

void
continuous_agg_invalidate_mat_ht(const Hypertable *raw_ht, const Hypertable *mat_ht, int64 start,
								 int64 end)
{
	Assert((raw_ht != NULL) && (mat_ht != NULL));

	if (hypertable_is_distributed(raw_ht))
	{
		remote_invalidation_log_add_entry(raw_ht,
										  HypertableIsMaterialization,
										  mat_ht->fd.id,
										  start,
										  end);
	}
	else
	{
		invalidation_cagg_log_add_entry(mat_ht->fd.id, start, end);
	}
}

typedef enum InvalidationResult
{
	INVAL_NOMATCH,
	INVAL_DELETE,
	INVAL_CUT,
} InvalidationResult;

#define IS_VALID_INVALIDATION(entry) ((entry)->hyper_id > 0)

static bool
save_invalidation_for_refresh(const CaggInvalidationState *state, const Invalidation *invalidation)
{
	int32 cagg_hyper_id = state->mat_hypertable_id;
	TupleDesc tupdesc = RelationGetDescr(state->cagg_log_rel);
	HeapTuple refresh_tup;

	if (!IS_VALID_INVALIDATION(invalidation))
		return false;

	refresh_tup = create_invalidation_tup(tupdesc,
										  cagg_hyper_id,
										  invalidation->lowest_modified_value,
										  invalidation->greatest_modified_value);
	tuplestore_puttuple(state->invalidations, refresh_tup);
	heap_freetuple(refresh_tup);

	return true;
}

static void
set_remainder_after_cut(Invalidation *remainder, int32 hyper_id, int64 lowest_modified_value,
						int64 greatest_modified_value)
{
	MemSet(remainder, 0, sizeof(*remainder));
	remainder->hyper_id = hyper_id;
	remainder->lowest_modified_value = lowest_modified_value;
	remainder->greatest_modified_value = greatest_modified_value;
}

/*
 * Try to cut an invalidation against the refresh window.
 *
 * If an invalidation entry overlaps with the refresh window, it needs
 * additional processing: it is either cut, deleted, or left unmodified.
 *
 * The part(s) of the invalidation that are outside the refresh window after
 * the cut will remain in the log. The part of the invalidation that fits
 * within the window is returned as the "remainder".
 *
 * Note that the refresh window is exclusive in the end while invalidations
 * are inclusive.
 */
static InvalidationResult
cut_invalidation_along_refresh_window(const CaggInvalidationState *state,
									  const Invalidation *invalidation,
									  const InternalTimeRange *refresh_window,
									  Invalidation *remainder)
{
	int32 cagg_hyper_id = state->mat_hypertable_id;
	TupleDesc tupdesc = RelationGetDescr(state->cagg_log_rel);
	InvalidationResult result = INVAL_NOMATCH;
	HeapTuple lower = NULL;
	HeapTuple upper = NULL;

	Assert(remainder != NULL);

	/* Entry is completely enclosed by the refresh window */
	if (invalidation->lowest_modified_value >= refresh_window->start &&
		invalidation->greatest_modified_value < refresh_window->end)
	{
		/*
		 * Entry completely enclosed so can be deleted:
		 *
		 * [---------------)
		 *     [+++++]
		 */

		result = INVAL_DELETE;
		set_remainder_after_cut(remainder,
								cagg_hyper_id,
								invalidation->lowest_modified_value,
								invalidation->greatest_modified_value);
	}
	else
	{
		if (invalidation->lowest_modified_value < refresh_window->start &&
			invalidation->greatest_modified_value >= refresh_window->start)
		{
			/*
			 * Need to cut in right end:
			 *
			 *     [------)
			 * [++++++]
			 *
			 * [++]
			 */
			lower = create_invalidation_tup(tupdesc,
											cagg_hyper_id,
											invalidation->lowest_modified_value,
											refresh_window->start - 1);
			set_remainder_after_cut(remainder,
									cagg_hyper_id,
									refresh_window->start,
									/* Refresh window not exclusive at end */
									MIN(refresh_window->end - 1,
										invalidation->greatest_modified_value));
			result = INVAL_CUT;
		}

		if (invalidation->lowest_modified_value < refresh_window->end &&
			invalidation->greatest_modified_value >= refresh_window->end)
		{
			/*
			 * If the invalidation is already cut on the left above, the reminder is set and
			 * will be reset here. The assert prevents from losing information from the reminder.
			 */
			Assert((result == INVAL_CUT &&
					remainder->lowest_modified_value == refresh_window->start) ||
				   result == INVAL_NOMATCH);

			/*
			 * Need to cut in left end:
			 *
			 * [------)
			 *    [++++++++]
			 *
			 *        [++++]
			 */
			upper = create_invalidation_tup(tupdesc,
											cagg_hyper_id,
											refresh_window->end,
											invalidation->greatest_modified_value);
			set_remainder_after_cut(remainder,
									cagg_hyper_id,
									MAX(invalidation->lowest_modified_value, refresh_window->start),
									/* Refresh window exclusive at end */
									refresh_window->end - 1);
			result = INVAL_CUT;
		}
	}

	/* Insert any modifications into the cagg invalidation log */
	if (result == INVAL_CUT)
	{
		CatalogSecurityContext sec_ctx;
		HeapTuple other_range = NULL;

		ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);

		/* We'd like to do one update (unless the TID is not set), and
		 * optionally one insert. We pick one of the tuples for an update, and
		 * the other one will be an insert. */
		if (lower || upper)
		{
			HeapTuple tup = lower ? lower : upper;
			other_range = lower ? upper : lower;

			/* If the TID is set, we are updating an existing tuple, i.e., we
			 * are processing and entry in the cagg log itself. Otherwise, we
			 * are processing the hypertable invalidation log and need to
			 * insert a new entry. */
			if (ItemPointerIsValid(&invalidation->tid))
			{
				ItemPointerData tid = invalidation->tid;
				ts_catalog_update_tid_only(state->cagg_log_rel, &tid, tup);
			}
			else
				ts_catalog_insert_only(state->cagg_log_rel, tup);

			heap_freetuple(tup);
		}

		if (other_range)
		{
			ts_catalog_insert_only(state->cagg_log_rel, other_range);
			heap_freetuple(other_range);
		}

		ts_catalog_restore_user(&sec_ctx);
	}

	return result;
}

static void
invalidation_entry_reset(Invalidation *entry)
{
	MemSet(entry, 0, sizeof(Invalidation));
}

/*
 * Expand an invalidation to bucket boundaries.
 *
 * Since a refresh always materializes full buckets, we can safely expand an
 * invalidation to bucket boundaries and in the process merge a lot more
 * invalidations.
 */
static void
invalidation_expand_to_bucket_boundaries(Invalidation *inv, Oid time_type_oid, int64 bucket_width,
										 const ContinuousAggsBucketFunction *bucket_function)
{
	const int64 time_dimension_min = ts_time_get_min(time_type_oid);
	const int64 time_dimension_max = ts_time_get_max(time_type_oid);
	int64 min_bucket_start;
	int64 max_bucket_end;

	if (bucket_width == BUCKET_WIDTH_VARIABLE)
	{
		ts_compute_circumscribed_bucketed_refresh_window_variable(&inv->lowest_modified_value,
																  &inv->greatest_modified_value,
																  bucket_function);
		return;
	}

	Assert(bucket_width > 0);

	/* Compute the start of the "first" bucket for the type. The min value
	 * must be at the start of the "first" bucket or somewhere in the
	 * bucket. If the min value falls on the exact start of the bucket we are
	 * good. Otherwise, we need to move to the next full bucket. */
	min_bucket_start = ts_time_saturating_add(time_dimension_min, bucket_width - 1, time_type_oid);
	min_bucket_start = ts_time_bucket_by_type(bucket_width, min_bucket_start, time_type_oid);

	/* Compute the end of the "last" bucket for the time type. Remember that
	 * invalidations are inclusive, so the "greatest" value should be the last
	 * value of the last full bucket. Either the max value is already the last
	 * value of the last bucket, or we need to return the last value of the
	 * previous full bucket.  */
	max_bucket_end = ts_time_bucket_by_type(bucket_width, time_dimension_max, time_type_oid);

	/* Check if the max value was already the last value of the last bucket */
	if (ts_time_saturating_add(max_bucket_end, bucket_width - 1, time_type_oid) ==
		time_dimension_max)
	{
		max_bucket_end = time_dimension_max;
	}
	else
	{
		/* The last bucket was partial. To get the end of previous bucket, we
		 * need to move one step down from the partial last bucket. */
		max_bucket_end = ts_time_saturating_sub(max_bucket_end, 1, time_type_oid);
	}

	if (inv->lowest_modified_value < min_bucket_start)
		/* Below the min bucket, so treat as invalid to -infinity. */
		inv->lowest_modified_value = INVAL_NEG_INFINITY;
	else if (inv->lowest_modified_value > max_bucket_end)
		/* Above the max bucket, so treat as invalid to +infinity. */
		inv->lowest_modified_value = INVAL_POS_INFINITY;
	else
		inv->lowest_modified_value =
			ts_time_bucket_by_type(bucket_width, inv->lowest_modified_value, time_type_oid);

	if (inv->greatest_modified_value < min_bucket_start)
		/* Below the min bucket, so treat as invalid to -infinity. */
		inv->greatest_modified_value = INVAL_NEG_INFINITY;
	else if (inv->greatest_modified_value > max_bucket_end)
		/* Above the max bucket, so treat as invalid to +infinity. */
		inv->greatest_modified_value = INVAL_POS_INFINITY;
	else
	{
		inv->greatest_modified_value =
			ts_time_bucket_by_type(bucket_width, inv->greatest_modified_value, time_type_oid);
		inv->greatest_modified_value =
			ts_time_saturating_add(inv->greatest_modified_value, bucket_width - 1, time_type_oid);
	}
}

/*
 * Macro to set an Invalidation from a tuple. The tuple can either have the
 * format of the hypertable invalidation log or the continuous aggregate
 * invalidation log (as determined by the type parameter).
 */
#define INVALIDATION_ENTRY_SET(entry, ti, hypertable_id, type)                                     \
	do                                                                                             \
	{                                                                                              \
		bool should_free;                                                                          \
		HeapTuple tuple = ts_scanner_fetch_heap_tuple(ti, false, &should_free);                    \
		type form;                                                                                 \
		form = (type) GETSTRUCT(tuple);                                                            \
		(entry)->hyper_id = form->hypertable_id;                                                   \
		(entry)->lowest_modified_value = form->lowest_modified_value;                              \
		(entry)->greatest_modified_value = form->greatest_modified_value;                          \
		(entry)->is_modified = false;                                                              \
		ItemPointerCopy(&tuple->t_self, &(entry)->tid);                                            \
                                                                                                   \
		if (should_free)                                                                           \
			heap_freetuple(tuple);                                                                 \
	} while (0);

static void
invalidation_entry_set_from_hyper_invalidation(Invalidation *entry, const TupleInfo *ti,
											   int32 hyper_id, Oid dimtype, int64 bucket_width,
											   const ContinuousAggsBucketFunction *bucket_function)
{
	INVALIDATION_ENTRY_SET(entry,
						   ti,
						   hypertable_id,
						   Form_continuous_aggs_hypertable_invalidation_log);
	/* Since hypertable invalidations are moved to the continuous aggregate
	 * invalidation log, a different hypertable ID must be set (the ID of the
	 * materialized hypertable). */
	entry->hyper_id = hyper_id;
	invalidation_expand_to_bucket_boundaries(entry, dimtype, bucket_width, bucket_function);
}

static void
invalidation_entry_set_from_cagg_invalidation(Invalidation *entry, const TupleInfo *ti, Oid dimtype,
											  int64 bucket_width,
											  const ContinuousAggsBucketFunction *bucket_function)
{
	INVALIDATION_ENTRY_SET(entry,
						   ti,
						   materialization_id,
						   Form_continuous_aggs_materialization_invalidation_log);

	/* It isn't strictly necessary to expand the invalidation to bucket
	 * boundaries here since all invalidations were already expanded when
	 * copied from the hypertable invalidation log. However, since
	 * invalidation expansion wasn't implemented in early 2.0.x versions of
	 * the extension, there might be unexpanded entries in the cagg
	 * invalidation log for some users. Therefore we try to expand
	 * invalidation entries also here, although in most cases it would do
	 * nothing. */
	invalidation_expand_to_bucket_boundaries(entry, dimtype, bucket_width, bucket_function);
}

/*
 * Check if two invalidations can be merged into one.
 *
 * Since invalidations are inclusive in both ends, two adjacent invalidations
 * can be merged.
 */
static bool
invalidations_can_be_merged(const Invalidation *a, const Invalidation *b)
{
	/* To account for adjacency, expand one window 1 step in each
	 * direction. This makes adjacent invalidations overlapping. */
	int64 a_start = int64_saturating_sub(a->lowest_modified_value, 1);
	int64 a_end = int64_saturating_add(a->greatest_modified_value, 1);

	return a_end >= b->lowest_modified_value && a_start <= b->greatest_modified_value;
}

/*
 * Try to merge two invalidations into one.
 *
 * Returns true if the invalidations were merged, otherwise false.
 *
 * Given that we scan ordered on lowest_modified_value, the previous and
 * current invalidation can overlap in two ways (generalized):
 *
 * |------|
 *    |++++++++|
 *
 * |-------------|
 *    |++++++++|
 *
 * The closest non-overlapping case is (note that adjacent invalidations can
 * be merged since they are inclusive in both ends):
 *
 * |--|
 *     |++++++++|
 *
 */
static bool
invalidation_entry_try_merge(Invalidation *entry, const Invalidation *newentry)
{
	if (!IS_VALID_INVALIDATION(newentry))
		return false;

	/* Quick exit if no overlap */
	if (!invalidations_can_be_merged(entry, newentry))
		return false;

	/* Check if the new entry expands beyond the old one (first case above) */
	if (entry->greatest_modified_value < newentry->greatest_modified_value)
	{
		entry->greatest_modified_value = newentry->greatest_modified_value;
		entry->is_modified = true;
	}

	return true;
}

static void
cut_and_insert_new_cagg_invalidation(const CaggInvalidationState *state, const Invalidation *entry,
									 int32 cagg_hyper_id)
{
	CatalogSecurityContext sec_ctx;
	TupleDesc tupdesc = RelationGetDescr(state->cagg_log_rel);
	HeapTuple newtup;

	newtup = create_invalidation_tup(tupdesc,
									 cagg_hyper_id,
									 entry->lowest_modified_value,
									 entry->greatest_modified_value);

	ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
	ts_catalog_insert_only(state->cagg_log_rel, newtup);
	ts_catalog_restore_user(&sec_ctx);
}

/*
 * Process invalidations in the hypertable invalidation log.
 *
 * Copy and delete all entries from the hypertable invalidation log.
 *
 * Note that each entry gets one copy per continuous aggregate in the cagg
 * invalidation log (unless it was merged or matched the refresh
 * window). These copied entries are later used to track invalidations across
 * refreshes on a per-cagg basis.
 *
 * After this function has run, there are no entries left in the hypertable
 * invalidation log.
 */
static void
move_invalidations_from_hyper_to_cagg_log(const CaggInvalidationState *state)
{
	const CaggsInfo *all_caggs = state->all_caggs;
	int32 hyper_id = state->raw_hypertable_id;
	int32 last_cagg_hyper_id;
	ListCell *lc1, *lc2, *lc3;

	last_cagg_hyper_id = llast_int(all_caggs->mat_hypertable_ids);

	/* We use a per-tuple memory context in the scan loop since we could be
	 * processing a lot of invalidations (basically an unbounded
	 * amount). Initialize it here by resetting it. */
	MemoryContextReset(state->per_tuple_mctx);

	/*
	 * Looping over all continuous aggregates in the outer loop ensures all
	 * tuples for a specific continuous aggregate is inserted consecutively in
	 * the cagg invalidation log. This creates better locality for scanning
	 * the invalidations later.
	 */
	forthree (lc1,
			  all_caggs->mat_hypertable_ids,
			  lc2,
			  all_caggs->bucket_widths,
			  lc3,
			  all_caggs->bucket_functions)
	{
		int32 cagg_hyper_id = lfirst_int(lc1);
		int64 bucket_width = DatumGetInt64(PointerGetDatum(lfirst(lc2)));
		const ContinuousAggsBucketFunction *bucket_function = lfirst(lc3);

		Invalidation mergedentry;
		ScanIterator iterator;

		invalidation_entry_reset(&mergedentry);
		hypertable_invalidation_scan_init(&iterator, hyper_id, RowExclusiveLock);
		iterator.ctx.snapshot = state->snapshot;

		/* Scan all invalidations */
		ts_scanner_foreach(&iterator)
		{
			TupleInfo *ti;
			MemoryContext oldmctx;
			Invalidation logentry;

			oldmctx = MemoryContextSwitchTo(state->per_tuple_mctx);
			ti = ts_scan_iterator_tuple_info(&iterator);

			invalidation_entry_set_from_hyper_invalidation(&logentry,
														   ti,
														   cagg_hyper_id,
														   state->dimtype,
														   bucket_width,
														   bucket_function);

			if (!IS_VALID_INVALIDATION(&mergedentry))
			{
				mergedentry = logentry;
				mergedentry.hyper_id = cagg_hyper_id;
			}
			else if (!invalidation_entry_try_merge(&mergedentry, &logentry))
			{
				cut_and_insert_new_cagg_invalidation(state, &mergedentry, cagg_hyper_id);
				mergedentry = logentry;
			}

			if (cagg_hyper_id == last_cagg_hyper_id)
			{
				CatalogSecurityContext sec_ctx;

				/* The invalidation has been processed for all caggs, so the
				 * only thing left is to delete it from the source hypertable
				 * invalidation log. */
				ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
				ts_catalog_delete_tid_only(ti->scanrel, &logentry.tid);
				ts_catalog_restore_user(&sec_ctx);
			}

			MemoryContextSwitchTo(oldmctx);
			MemoryContextReset(state->per_tuple_mctx);
		}

		ts_scan_iterator_close(&iterator);

		/* Handle the last merged invalidation */
		if (IS_VALID_INVALIDATION(&mergedentry))
			cut_and_insert_new_cagg_invalidation(state, &mergedentry, cagg_hyper_id);
	}
}

static void
cagg_invalidations_scan_by_hypertable_init(ScanIterator *iterator, int32 cagg_hyper_id,
										   LOCKMODE lockmode)
{
	*iterator = ts_scan_iterator_create(CONTINUOUS_AGGS_MATERIALIZATION_INVALIDATION_LOG,
										lockmode,
										CurrentMemoryContext);
	iterator->ctx.index = catalog_get_index(ts_catalog_get(),
											CONTINUOUS_AGGS_MATERIALIZATION_INVALIDATION_LOG,
											CONTINUOUS_AGGS_MATERIALIZATION_INVALIDATION_LOG_IDX);
	ts_scan_iterator_scan_key_init(
		iterator,
		Anum_continuous_aggs_materialization_invalidation_log_idx_materialization_id,
		BTEqualStrategyNumber,
		F_INT4EQ,
		Int32GetDatum(cagg_hyper_id));
}

/*
 * Cut an invalidation and return the part, if any, that remains within the
 * refresh window.
 */
static Invalidation
cut_cagg_invalidation(const CaggInvalidationState *state, const InternalTimeRange *refresh_window,
					  const Invalidation *entry)
{
	InvalidationResult result;
	Invalidation remainder;
	ItemPointerData tid = entry->tid;

	invalidation_entry_reset(&remainder);

	result = cut_invalidation_along_refresh_window(state, entry, refresh_window, &remainder);

	switch (result)
	{
		case INVAL_NOMATCH:
			/* If no cutting was done (i.e., the invalidation was outside the
			 * refresh window), but the invalidation was previously merged
			 * (expanded) with another invalidation, then we still need to
			 * update it. */
			if (entry->is_modified)
			{
				HeapTuple tuple = create_invalidation_tup(RelationGetDescr(state->cagg_log_rel),
														  entry->hyper_id,
														  entry->lowest_modified_value,
														  entry->greatest_modified_value);
				ts_catalog_update_tid_only(state->cagg_log_rel, &tid, tuple);
				heap_freetuple(tuple);
			}
			break;
		case INVAL_DELETE:
			ts_catalog_delete_tid_only(state->cagg_log_rel, &tid);
			break;
		case INVAL_CUT:
			/* Nothing to do */
			break;
	}

	return remainder;
}

static Invalidation
cut_cagg_invalidation_and_compute_remainder(const CaggInvalidationState *state,
											const InternalTimeRange *refresh_window,
											const Invalidation *mergedentry,
											const Invalidation *current_remainder)
{
	Invalidation new_remainder;
	Invalidation remainder = *current_remainder;

	/* The previous and current invalidation could not be merged. We
	 * need to cut the prev invalidation against the refresh window */
	new_remainder = cut_cagg_invalidation(state, refresh_window, mergedentry);

	if (!IS_VALID_INVALIDATION(&remainder))
		remainder = new_remainder;
	else if (!invalidation_entry_try_merge(&remainder, &new_remainder))
	{
		save_invalidation_for_refresh(state, &remainder);
		remainder = new_remainder;
	}

	return remainder;
}

/*
 * Clear all cagg invalidations that match a refresh window.
 *
 * This function clears all invalidations in the cagg invalidation log that
 * matches a window, and adds the invalidation segments covered by the window
 * to the invalidation store (tuple store) in the state argument. The
 * remaining segments that are added to the invalidation store are regions
 * that require materialization.
 *
 * An invalidation entry that gets processed is either completely enclosed
 * (covered) by the refresh window, or it partially overlaps. In the former
 * case, the invalidation entry is removed and for the latter case it is
 * cut. Thus, an entry can either disappear, reduce in size, or be cut in two.
 *
 * Note that the refresh window is inclusive at the start and exclusive at the
 * end. This function also assumes that invalidations are scanned in order of
 * lowest_modified_value.
 */
static void
clear_cagg_invalidations_for_refresh(const CaggInvalidationState *state,
									 const InternalTimeRange *refresh_window)
{
	ScanIterator iterator;
	int32 cagg_hyper_id = state->mat_hypertable_id;
	Invalidation mergedentry;
	Invalidation remainder;

	invalidation_entry_reset(&mergedentry);
	invalidation_entry_reset(&remainder);
	cagg_invalidations_scan_by_hypertable_init(&iterator, cagg_hyper_id, RowExclusiveLock);
	iterator.ctx.snapshot = state->snapshot;

	MemoryContextReset(state->per_tuple_mctx);

	/* Process all invalidations for the continuous aggregate */
	ts_scanner_foreach(&iterator)
	{
		TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
		MemoryContext oldmctx;
		Invalidation logentry;
		int64 bucket_width = state->bucket_width;
		const ContinuousAggsBucketFunction *bucket_function = state->bucket_function;

		oldmctx = MemoryContextSwitchTo(state->per_tuple_mctx);
		invalidation_entry_set_from_cagg_invalidation(&logentry,
													  ti,
													  state->dimtype,
													  bucket_width,
													  bucket_function);

		if (!IS_VALID_INVALIDATION(&mergedentry))
			mergedentry = logentry;
		else if (invalidation_entry_try_merge(&mergedentry, &logentry))
		{
			/*
			 * The previous and current invalidation were merged into
			 * one entry (i.e., they overlapped or were adjacent).
			 */
			ts_catalog_delete_tid_only(state->cagg_log_rel, &logentry.tid);
		}
		else
		{
			remainder = cut_cagg_invalidation_and_compute_remainder(state,
																	refresh_window,
																	&mergedentry,
																	&remainder);
			mergedentry = logentry;
		}

		MemoryContextSwitchTo(oldmctx);
		MemoryContextReset(state->per_tuple_mctx);
	}

	ts_scan_iterator_close(&iterator);

	/* Handle the last (merged) invalidation */
	if (IS_VALID_INVALIDATION(&mergedentry))
		remainder = cut_cagg_invalidation_and_compute_remainder(state,
																refresh_window,
																&mergedentry,
																&remainder);

	/* Handle the last (merged) remainder */
	save_invalidation_for_refresh(state, &remainder);
}

static void
invalidation_state_init(CaggInvalidationState *state, int32 mat_hypertable_id,
						int32 raw_hypertable_id, Oid dimtype, const CaggsInfo *all_caggs)
{
	ListCell *lc1, *lc2, *lc3;
	bool PG_USED_FOR_ASSERTS_ONLY found = false;

	state->mat_hypertable_id = mat_hypertable_id;
	state->raw_hypertable_id = raw_hypertable_id;
	state->dimtype = dimtype;
	state->all_caggs = all_caggs;
	state->cagg_log_rel = open_invalidation_log(LOG_CAGG, RowExclusiveLock);
	state->per_tuple_mctx = AllocSetContextCreate(CurrentMemoryContext,
												  "Continuous aggregate invalidations",
												  ALLOCSET_DEFAULT_SIZES);
	state->snapshot = RegisterSnapshot(GetTransactionSnapshot());
	forthree (lc1,
			  all_caggs->mat_hypertable_ids,
			  lc2,
			  all_caggs->bucket_widths,
			  lc3,
			  all_caggs->bucket_functions)
	{
		int32 cagg_hyper_id = lfirst_int(lc1);

		if (cagg_hyper_id == mat_hypertable_id)
		{
			state->bucket_width = DatumGetInt64(PointerGetDatum(lfirst(lc2)));
			state->bucket_function = lfirst(lc3);

			found = true;
			break;
		}
	}
	Assert(found);
}

static void
invalidation_state_cleanup(const CaggInvalidationState *state)
{
	table_close(state->cagg_log_rel, NoLock);
	UnregisterSnapshot(state->snapshot);
	MemoryContextDelete(state->per_tuple_mctx);
}

void
invalidation_process_hypertable_log(int32 mat_hypertable_id, int32 raw_hypertable_id, Oid dimtype,
									const CaggsInfo *all_caggs)
{
	CaggInvalidationState state;

	invalidation_state_init(&state, mat_hypertable_id, raw_hypertable_id, dimtype, all_caggs);
	move_invalidations_from_hyper_to_cagg_log(&state);
	invalidation_state_cleanup(&state);
}

/*
 * Generates the default bucket_functions[] argument for the following functions:
 *
 * - _timescaledb_internal.invalidation_process_hypertable_log()
 * - _timescaledb_internal.invalidation_process_cagg_log()
 *
 * Users are expected to have the same or higher version of TimescaleDB on the
 * data nodes. When the access node runs the older version that knows nothing
 * about the bucket_functions[] argument, the data nodes that runs newer version
 * will receive less arguments that they expect. When this happens data nodes
 * generate the default argument, and execute the rest of the logic as usual.
 */
static ArrayType *
bucket_functions_default_argument(int ndim)
{
	int i;
	Datum *bucketfunctions = palloc(sizeof(Datum) * ndim);

	for (i = 0; i < ndim; i++)
		bucketfunctions[i] = CStringGetTextDatum("");

	return construct_array(bucketfunctions, ndim, TEXTOID, -1, false, TYPALIGN_INT);
}

/**
 * Processes the hypertable invalidation log in a data node for all the CAGGs that belong to the
 * distributed hypertable with hypertable ID 'raw_hypertable_id' in the Access Node. The
 * invalidations are cut, merged and moved to the materialization invalidation log.
 *
 * @param mat_hypertable_id The hypertable ID of the CAGG materialized hypertable in the Access
 *                          Node that is currently being refreshed.
 * @param raw_hypertable_id The hypertable ID of the original distributed hypertable in the Access
 *                          Node.
 * @param dimtype The OID of the type of the time dimension for this CAGG.
 * @param mat_hypertable_ids The array of hypertable IDs for all CAGG materialized hypertables in
 *                           the Access Node that belong to 'raw_hypertable_id'.
 * @param bucket_widths The array of time bucket widths for all the CAGGs that belong to
 *                      'raw_hypertable_id'.
 * @param max_bucket_widths (Deprecated) This argument is ignored and is present only for backward
 *                          compatibility.
 * @param bucket_functions (Optional) The array of serialized information about bucket functions.
 */
Datum
tsl_invalidation_process_hypertable_log(PG_FUNCTION_ARGS)
{
	int32 mat_hypertable_id = PG_GETARG_INT32(0);
	int32 raw_hypertable_id = PG_GETARG_INT32(1);
	Oid dimtype = PG_GETARG_OID(2);
	ArrayType *mat_hypertable_ids = PG_GETARG_ARRAYTYPE_P(3);
	ArrayType *bucket_widths = PG_GETARG_ARRAYTYPE_P(4);
	ArrayType *bucket_functions = PG_NARGS() > 6 ?
									  PG_GETARG_ARRAYTYPE_P(6) :
									  bucket_functions_default_argument(ARR_NDIM(bucket_widths));
	CaggsInfo all_caggs_info;

	ts_populate_caggs_info_from_arrays(mat_hypertable_ids,
									   bucket_widths,
									   bucket_functions,
									   &all_caggs_info);

	invalidation_process_hypertable_log(mat_hypertable_id,
										raw_hypertable_id,
										dimtype,
										&all_caggs_info);
	PG_RETURN_VOID();
}

#define INVALIDATION_PROCESS_HYPERTABLE_LOG_NARGS 7
#define INVALIDATION_PROCESS_HYPERTABLE_LOG_FUNCNAME "invalidation_process_hypertable_log"

void
remote_invalidation_process_hypertable_log(int32 mat_hypertable_id, int32 raw_hypertable_id,
										   Oid dimtype, const CaggsInfo *all_caggs)
{
	Oid func_oid;
	ArrayType *mat_hypertable_ids;
	ArrayType *bucket_widths;
	ArrayType *bucket_functions;
	LOCAL_FCINFO(fcinfo, INVALIDATION_PROCESS_HYPERTABLE_LOG_NARGS);
	FmgrInfo flinfo;
	unsigned int i;

	ts_create_arrays_from_caggs_info(all_caggs,
									 &mat_hypertable_ids,
									 &bucket_widths,
									 &bucket_functions);

	static const Oid type_id[INVALIDATION_PROCESS_HYPERTABLE_LOG_NARGS] = {
		INT4OID, INT4OID, REGTYPEOID, INT4ARRAYOID, INT8ARRAYOID, INT8ARRAYOID, TEXTARRAYOID
	};
	List *const fqn = list_make2(makeString(INTERNAL_SCHEMA_NAME),
								 makeString(INVALIDATION_PROCESS_HYPERTABLE_LOG_FUNCNAME));

	/*
	 * Note that we have to explicitly specify the amount of arguments in this
	 * case, because there are several overloaded versions of invalidation_process_hypertable_log().
	 */
	func_oid = LookupFuncName(fqn, lengthof(type_id), type_id, false);
	Assert(OidIsValid(func_oid));

	fmgr_info(func_oid, &flinfo);
	InitFunctionCallInfoData(*fcinfo,
							 &flinfo,
							 INVALIDATION_PROCESS_HYPERTABLE_LOG_NARGS,
							 InvalidOid,
							 NULL,
							 NULL);

	for (i = 0; i < INVALIDATION_PROCESS_HYPERTABLE_LOG_NARGS; ++i)
	{
		FC_NULL(fcinfo, i) = false;
	}
	FC_ARG(fcinfo, 0) = Int32GetDatum(mat_hypertable_id);
	FC_ARG(fcinfo, 1) = Int32GetDatum(raw_hypertable_id);
	FC_ARG(fcinfo, 2) = ObjectIdGetDatum(dimtype);
	FC_ARG(fcinfo, 3) = PointerGetDatum(mat_hypertable_ids);
	FC_ARG(fcinfo, 4) = PointerGetDatum(bucket_widths);
	FC_ARG(fcinfo, 5) = PointerGetDatum(construct_empty_array(INT8OID));
	FC_ARG(fcinfo, 6) = PointerGetDatum(bucket_functions);
	/* Check for null result, since caller is clearly not expecting one */
	if (fcinfo->isnull)
		elog(ERROR, "function %u returned NULL", flinfo.fn_oid);

	Hypertable *ht = ts_hypertable_get_by_id(raw_hypertable_id);
	if (!ht || !hypertable_is_distributed(ht))
		elog(ERROR, "function was not provided with a valid distributed hypertable id");

	DistCmdResult *result;
	List *data_node_list = ts_hypertable_get_data_node_name_list(ht);
	result = ts_dist_cmd_invoke_func_call_on_data_nodes(fcinfo, data_node_list);
	if (result)
		ts_dist_cmd_close_response(result);
}

InvalidationStore *
invalidation_process_cagg_log(int32 mat_hypertable_id, int32 raw_hypertable_id,
							  const InternalTimeRange *refresh_window,
							  const CaggsInfo *all_caggs_info, const long max_materializations,
							  bool *do_merged_refresh, InternalTimeRange *ret_merged_refresh_window)
{
	CaggInvalidationState state;
	InvalidationStore *store = NULL;
	long count;

	*do_merged_refresh = false;

	invalidation_state_init(&state,
							mat_hypertable_id,
							raw_hypertable_id,
							refresh_window->type,
							all_caggs_info);
	state.invalidations = tuplestore_begin_heap(false, false, work_mem);
	clear_cagg_invalidations_for_refresh(&state, refresh_window);
	count = tuplestore_tuple_count(state.invalidations);

	if (count == 0)
	{
		tuplestore_end(state.invalidations);
	}
	else
	{
		store = palloc(sizeof(InvalidationStore));
		store->tupstore = state.invalidations;
		store->tupdesc = CreateTupleDescCopy(RelationGetDescr(state.cagg_log_rel));
	}

	invalidation_state_cleanup(&state);

	/*
	 * If there are many individual invalidation ranges to refresh, then
	 * revert to a merged refresh across the range decided by lowest and
	 * highest invalidated value.
	 */
	if (count && tuplestore_tuple_count(store->tupstore) > max_materializations)
	{
		InternalTimeRange merged_refresh_window;
		continuous_agg_calculate_merged_refresh_window(refresh_window,
													   store,
													   state.bucket_width,
													   state.bucket_function,
													   &merged_refresh_window);
		*do_merged_refresh = true;
		*ret_merged_refresh_window = merged_refresh_window;
		invalidation_store_free(store);
		store = NULL;
	}

	return store;
}

/**
 * Processes the materialization invalidation log in a data node for the CAGG being refreshed that
 * belongs to the distributed hypertable with hypertable ID 'raw_hypertable_id' in the Access Node.
 * The invalidations are cut, merged and returned as a single refresh window.
 *
 * @param mat_hypertable_id The hypertable ID of the CAGG materialized hypertable in the Access
 *                          Node that is currently being refreshed.
 * @param raw_hypertable_id The hypertable ID of the original distributed hypertable in the Access
 *                          Node.
 * @param dimtype The OID of the type of the time dimension for this CAGG.
 * @param window_start The starting time of the CAGG refresh window.
 * @param window_end The ending time of the CAGG refresh window.
 * @param mat_hypertable_ids The array of hypertable IDs for all CAGG materialized hypertables in
 *                           the Access Node that belong to 'raw_hypertable_id'.
 * @param bucket_widths The array of time bucket widths for all the CAGGs that belong to
 *                      'raw_hypertable_id'.
 * @param max_bucket_widths (Deprecated) This argument is ignored and is present only for backward
 *                          compatibility.
 * @param bucket_functions (Optional) The array of serialized information about bucket functions.
 * @return a tuple of:
 *         ret_window_start - The merged refresh window starting time
 *         ret_window_end - The merged refresh window ending time
 */
Datum
tsl_invalidation_process_cagg_log(PG_FUNCTION_ARGS)
{
	bool do_merged_refresh;
	InvalidationStore *invalidations pg_attribute_unused() = NULL;

	int32 mat_hypertable_id = PG_GETARG_INT32(0);
	int32 raw_hypertable_id = PG_GETARG_INT32(1);
	InternalTimeRange refresh_window = { .type = PG_GETARG_OID(2),
										 .start = PG_GETARG_INT64(3),
										 .end = PG_GETARG_INT64(4) };
	ArrayType *mat_hypertable_ids = PG_GETARG_ARRAYTYPE_P(5);
	ArrayType *bucket_widths = PG_GETARG_ARRAYTYPE_P(6);
	ArrayType *bucket_functions = PG_NARGS() > 8 ?
									  PG_GETARG_ARRAYTYPE_P(8) :
									  bucket_functions_default_argument(ARR_NDIM(bucket_widths));
	CaggsInfo all_caggs_info;
	InternalTimeRange ret_merged_refresh_window;

	ts_populate_caggs_info_from_arrays(mat_hypertable_ids,
									   bucket_widths,
									   bucket_functions,
									   &all_caggs_info);

	/* Force to always merge the refresh ranges since it is running in the data node
	 * by setting (max_materializations = 0) */
	invalidations = invalidation_process_cagg_log(mat_hypertable_id,
												  raw_hypertable_id,
												  &refresh_window,
												  &all_caggs_info,
												  0 /* max_materializations */,
												  &do_merged_refresh,
												  &ret_merged_refresh_window);
	HeapTuple tuple;
	TupleDesc tupdesc;
	Datum values[2];
	bool nulls[2] = { false, false };

	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "function returning record called in context that cannot accept type record");

	tupdesc = BlessTupleDesc(tupdesc);

	if (!do_merged_refresh)
	{ /* No invalidations found whatsoever */
		Assert(!invalidations);
		nulls[0] = true;
		nulls[1] = true;
	}
	else
	{
		values[0] = Int64GetDatum(ret_merged_refresh_window.start);
		values[1] = Int64GetDatum(ret_merged_refresh_window.end);
	}

	tuple = heap_form_tuple(tupdesc, values, nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(tuple));
}

#define INVALIDATION_PROCESS_CAGG_LOG_NARGS 9
#define INVALIDATION_PROCESS_CAGG_LOG_FUNCNAME "invalidation_process_cagg_log"

void
remote_invalidation_process_cagg_log(int32 mat_hypertable_id, int32 raw_hypertable_id,
									 const InternalTimeRange *refresh_window,
									 const CaggsInfo *all_caggs, bool *do_merged_refresh,
									 InternalTimeRange *ret_merged_refresh_window)
{
	Oid func_oid;
	ArrayType *mat_hypertable_ids;
	ArrayType *bucket_widths;
	ArrayType *bucket_functions;
	LOCAL_FCINFO(fcinfo, INVALIDATION_PROCESS_CAGG_LOG_NARGS);
	FmgrInfo flinfo;
	unsigned int i;

	*do_merged_refresh = false;

	ts_create_arrays_from_caggs_info(all_caggs,
									 &mat_hypertable_ids,
									 &bucket_widths,
									 &bucket_functions);

	static const Oid type_id[INVALIDATION_PROCESS_CAGG_LOG_NARGS] = {
		INT4OID,	  INT4OID,		REGTYPEOID,	  INT8OID,		INT8OID,
		INT4ARRAYOID, INT8ARRAYOID, INT8ARRAYOID, TEXTARRAYOID,
	};
	List *const fqn = list_make2(makeString(INTERNAL_SCHEMA_NAME),
								 makeString(INVALIDATION_PROCESS_CAGG_LOG_FUNCNAME));

	/*
	 * Note that we have to explicitly specify the amount of arguments in this
	 * case, because there are several overloaded versions of invalidation_process_cagg_log().
	 */
	func_oid = LookupFuncName(fqn, lengthof(type_id), type_id, false);
	Assert(OidIsValid(func_oid));

	fmgr_info(func_oid, &flinfo);
	InitFunctionCallInfoData(*fcinfo,
							 &flinfo,
							 INVALIDATION_PROCESS_CAGG_LOG_NARGS,
							 InvalidOid,
							 NULL,
							 NULL);

	for (i = 0; i < INVALIDATION_PROCESS_CAGG_LOG_NARGS; ++i)
	{
		FC_NULL(fcinfo, i) = false;
	}
	FC_ARG(fcinfo, 0) = Int32GetDatum(mat_hypertable_id);
	FC_ARG(fcinfo, 1) = Int32GetDatum(raw_hypertable_id);
	FC_ARG(fcinfo, 2) = ObjectIdGetDatum(refresh_window->type);
	FC_ARG(fcinfo, 3) = Int64GetDatum(refresh_window->start);
	FC_ARG(fcinfo, 4) = Int64GetDatum(refresh_window->end);
	FC_ARG(fcinfo, 5) = PointerGetDatum(mat_hypertable_ids);
	FC_ARG(fcinfo, 6) = PointerGetDatum(bucket_widths);
	FC_ARG(fcinfo, 7) = PointerGetDatum(construct_empty_array(INT8OID));
	FC_ARG(fcinfo, 8) = PointerGetDatum(bucket_functions);
	/* Check for null result, since caller is clearly not expecting one */
	if (fcinfo->isnull)
		elog(ERROR, "function %u returned NULL", flinfo.fn_oid);

	Hypertable *ht = ts_hypertable_get_by_id(raw_hypertable_id);
	if (!ht || !hypertable_is_distributed(ht))
		elog(ERROR, "function was not provided with a valid distributed hypertable id");

	DistCmdResult *dist_res;
	List *data_node_list = ts_hypertable_get_data_node_name_list(ht);
	dist_res = ts_dist_cmd_invoke_func_call_on_data_nodes(fcinfo, data_node_list);
	if (dist_res)
	{
		unsigned num_dist_res = ts_dist_cmd_response_count(dist_res);
		int64 start_time, end_time;
		InternalTimeRange merged_window = {
			.type = refresh_window->type,
			.start = TS_TIME_NOEND, /* initial state invalid */
			.end = TS_TIME_NOBEGIN	/* initial state invalid */
		};

		for (i = 0; i < num_dist_res; ++i)
		{
			const char *node_name;
			PGresult *result = ts_dist_cmd_get_result_by_index(dist_res, i, &node_name);
			if (PQresultStatus(result) != PGRES_TUPLES_OK)
				ereport(ERROR,
						(errcode(ERRCODE_CONNECTION_EXCEPTION),
						 errmsg("%s", PQresultErrorMessage(result))));
			Assert(PQntuples(result) == 1);
			Assert(PQnfields(result) == 2);
			if (PQgetisnull(result, 0, 0))
			{ /* No invalidations in this data node */
				Assert(PQgetisnull(result, 0, 1));
				continue;
			}
			start_time = pg_strtoint64(PQgetvalue(result, 0, 0));
			end_time = pg_strtoint64(PQgetvalue(result, 0, 1));
			elog(DEBUG1,
				 "merged invalidations for refresh on [" INT64_FORMAT ", " INT64_FORMAT "] from %s",
				 start_time,
				 end_time,
				 node_name);

			/* merge refresh windows from the data nodes */
			if (start_time < merged_window.start)
				merged_window.start = start_time;
			if (end_time > merged_window.end)
				merged_window.end = end_time;
		}
		ts_dist_cmd_close_response(dist_res);

		if (merged_window.start <= merged_window.end)
		{
			*ret_merged_refresh_window = merged_window;
			*do_merged_refresh = true;
		}
	}
}

#define INVALIDATION_LOG_DELETE_NARGS 1
#define HYPERTABLE_INVALIDATION_LOG_DELETE_FUNCNAME "hypertable_invalidation_log_delete"
#define MATERIALIZATION_INVALIDATION_LOG_DELETE_FUNCNAME "materialization_invalidation_log_delete"

void
remote_invalidation_log_delete(int32 raw_hypertable_id, ContinuousAggHypertableStatus caggstatus)
{
	/* Execute on all data nodes if there are any */
	List *data_nodes = data_node_get_node_name_list();
	if (NIL == data_nodes)
		return;

	Oid func_oid;
	LOCAL_FCINFO(fcinfo, INVALIDATION_LOG_DELETE_NARGS);
	FmgrInfo flinfo;

	Assert(HypertableIsMaterialization == caggstatus || HypertableIsRawTable == caggstatus);

	static const Oid type_id[INVALIDATION_LOG_DELETE_NARGS] = { INT4OID };
	List *const fqn = list_make2(makeString(INTERNAL_SCHEMA_NAME),
								 makeString((caggstatus == HypertableIsMaterialization) ?
												MATERIALIZATION_INVALIDATION_LOG_DELETE_FUNCNAME :
												HYPERTABLE_INVALIDATION_LOG_DELETE_FUNCNAME));

	func_oid = LookupFuncName(fqn, -1 /* lengthof(type_id) */, type_id, false);
	Assert(OidIsValid(func_oid));

	fmgr_info(func_oid, &flinfo);
	InitFunctionCallInfoData(*fcinfo,
							 &flinfo,
							 INVALIDATION_LOG_DELETE_NARGS,
							 InvalidOid,
							 NULL,
							 NULL);

	FC_NULL(fcinfo, 0) = false;
	FC_ARG(fcinfo, 0) = Int32GetDatum(raw_hypertable_id);
	/* Check for null result, since caller is clearly not expecting one */
	if (fcinfo->isnull)
		elog(ERROR, "function %u returned NULL", flinfo.fn_oid);

	DistCmdResult *result;
	result = ts_dist_cmd_invoke_func_call_on_data_nodes(fcinfo, data_nodes);
	if (result)
		ts_dist_cmd_close_response(result);
}

/**
 * Delete invalidation trigger for distributed hypertable member with hypertable ID
 * 'raw_hypertable_id' in the Data Node.
 *
 * @param raw_hypertable_id - The hypertable ID of the distributed hypertable member in the
 *                            Data Node.
 */
Datum
tsl_drop_dist_ht_invalidation_trigger(PG_FUNCTION_ARGS)
{
	Cache *hcache;
	Hypertable *ht;
	int32 raw_hypertable_id = PG_GETARG_INT32(0);

	hcache = ts_hypertable_cache_pin();
	ht = ts_hypertable_cache_get_entry_by_id(hcache, raw_hypertable_id);
	if (!ht || !hypertable_is_distributed_member(ht))
		elog(ERROR, "function was not provided with a valid distributed hypertable id");

	ts_materialization_invalidation_log_delete_inner(raw_hypertable_id);
	ts_hypertable_drop_trigger(ht->main_table_relid, CAGGINVAL_TRIGGER_NAME);

	ts_cache_release(hcache);

	PG_RETURN_VOID();
}

#define DROP_DIST_HT_INVALIDATION_TRIGGER_NARGS 1
#define DROP_DIST_HT_INVALIDATION_TRIGGER_FUNCNAME "drop_dist_ht_invalidation_trigger"

void
remote_drop_dist_ht_invalidation_trigger(int32 raw_hypertable_id)
{
	Cache *hcache;
	Hypertable *ht;

	hcache = ts_hypertable_cache_pin();
	ht = ts_hypertable_cache_get_entry_by_id(hcache, raw_hypertable_id);
	Assert(ht);
	if (!hypertable_is_distributed(ht))
	{
		ts_cache_release(hcache);
		return;
	}
	List *data_node_list = ts_hypertable_get_data_node_name_list(ht);
	DistCmdResult *result;
	ListCell *cell;
	Oid func_oid;

	static const Oid type_id[DROP_DIST_HT_INVALIDATION_TRIGGER_NARGS] = { INT4OID };
	List *const fqn = list_make2(makeString(INTERNAL_SCHEMA_NAME),
								 makeString(DROP_DIST_HT_INVALIDATION_TRIGGER_FUNCNAME));

	func_oid = LookupFuncName(fqn, -1 /* lengthof(type_id) */, type_id, false);
	Assert(OidIsValid(func_oid));

	FunctionCallInfo fcinfo =
		palloc(SizeForFunctionCallInfo(DROP_DIST_HT_INVALIDATION_TRIGGER_NARGS));
	FmgrInfo flinfo;
	List *cmd_descriptors = NIL; /* same order as ht->data_nodes */
	DistCmdDescr *cmd_descr_data = palloc(list_length(data_node_list) * sizeof(*cmd_descr_data));
	unsigned i = 0;

	foreach (cell, ht->data_nodes)
	{
		HypertableDataNode *node = lfirst(cell);

		fmgr_info(func_oid, &flinfo);
		InitFunctionCallInfoData(*fcinfo,
								 &flinfo,
								 DROP_DIST_HT_INVALIDATION_TRIGGER_NARGS,
								 InvalidOid,
								 NULL,
								 NULL);

		FC_NULL(fcinfo, 0) = false;
		/* distributed member hypertable ID */
		FC_ARG(fcinfo, 0) = Int32GetDatum(node->fd.node_hypertable_id);
		/* Check for null result, since caller is clearly not expecting one */
		if (fcinfo->isnull)
			elog(ERROR, "function %u returned NULL", flinfo.fn_oid);

		cmd_descr_data[i].sql = deparse_func_call(fcinfo);
		cmd_descr_data[i].params = NULL;
		cmd_descriptors = lappend(cmd_descriptors, &cmd_descr_data[i++]);
	}

	result = ts_dist_multi_cmds_params_invoke_on_data_nodes(cmd_descriptors, data_node_list, true);
	if (result)
		ts_dist_cmd_close_response(result);
	ts_cache_release(hcache);
}

void
invalidation_store_free(InvalidationStore *store)
{
	FreeTupleDesc(store->tupdesc);
	tuplestore_end(store->tupstore);
	pfree(store);
}
